package com.atguigu.flink.tableapi;

import com.atguigu.flink.function.WaterSensorMapFunction;
import com.atguigu.flink.pojo.WaterSensor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.*;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import static org.apache.flink.table.api.Expressions.*;

/**
 * Created by Smexy on 2023/3/5
 */
public class Demo9_GroupWindow
{
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        env.setParallelism(1);

        WatermarkStrategy<WaterSensor> watermarkStrategy = WatermarkStrategy
            .<WaterSensor>forMonotonousTimestamps()
            .withTimestampAssigner( (e, ts) -> e.getTs());

        SingleOutputStreamOperator<WaterSensor> ds = env
            .socketTextStream("hadoop103", 8888)
            .map(new WaterSensorMapFunction())
            //在流中产生水印
            .assignTimestampsAndWatermarks(watermarkStrategy);

        Table table = tableEnv.fromDataStream(ds, $("id"),$("ts"),$("vc"),
            $("pt").proctime(),
            $("et").rowtime());


        /*
                GroupWindow:  通常在之前学习的window

                    Table table = input
                      .window([GroupWindow w].as("w"))  // define window with alias w
                      .groupBy($("w"))  // group the table by window w
                      .select($("b").sum());  // aggregate

                overWindow:  sql中通用的开窗运算
                                select  xx,  窗口函数() over( 定义窗口的属性 )
         */

        /*
            定义窗口
                基于处理时间，基于事件时间
                滚动，滑动，会话
                        Tumble ： 定义滚动窗口
                        Slide ： 定义滑动窗口
                        Session ： 定义会话窗口
                            都有以下的方法:
                                over：  定义窗口的size，可以是时间或元素个数
                                every：  只有滑动窗口需要指定，是滑动的slide，可以是时间或元素个数
                                on：     定义时间属性。 如果是基于时间的窗口，定义时间字段。
                                        基于元素个数的，也要指定一个时间，用来排序数据。
                                as：   指定别名，可以在后续的groupBy或select使用别名指向窗口。
                                withGap： session窗口指定 间隔时间
                基于个数运算，基于时间运算
         */

        //基于处理时间的窗口 无需水印
        TumbleWithSizeOnTimeWithAlias w1 = Tumble.over(lit(5).seconds()).on($("pt")).as("w");
        SlideWithSizeAndSlideOnTimeWithAlias w3 = Slide.over(lit(5).seconds()).every(lit(3).seconds()).on($("pt")).as("w");
        SessionWithGapOnTimeWithAlias w5 = Session.withGap(lit(3).seconds()).on($("pt")).as("w");

        //基于事件时间的窗口 定义水印
        TumbleWithSizeOnTimeWithAlias w2 = Tumble.over(lit(5).seconds()).on($("et")).as("w");
        SlideWithSizeAndSlideOnTimeWithAlias w4 = Slide.over(lit(5).seconds()).every(lit(3).seconds()).on($("et")).as("w");
        SessionWithGapOnTimeWithAlias w6 = Session.withGap(lit(3).seconds()).on($("et")).as("w");

        //基于个数的窗口  rowInterval(): 定义行的间隔
        TumbleWithSizeOnTimeWithAlias w7 = Tumble.over(rowInterval(3l)).on($("pt")).as("w");
        // sql中的滑动窗口，第一次滑动，必须满足 size，才会运算。
        SlideWithSizeAndSlideOnTimeWithAlias w8 = Slide.over(rowInterval(3l)).every(rowInterval(2l)).on($("pt")).as("w");

        //计算
        table
             .window(w8)
             //全局窗口  所有的key都可以进入同一个窗口
             //.groupBy($("w"))
             //keyed 窗口
             .groupBy($("w"),$("id"))
             //时间窗口，可以获取窗口的时间范围
             //.select($("w").start().as("starttime"),$("w").end().as("endtime"),$("id"),$("vc").sum().as("sumVC"))
             // 个数窗口，没有时间范围属性
             .select($("id"),$("vc").sum().as("sumVC"))
             .execute()
             .print();



    }
}
